Usenet 1993 July/InfoMagic USENET CD-ROM July 1993.ISO/sources/unix/volume22/queuer/part03

Internet Message Format | 1990-06-07 | 33.4 KB

Subject:  v22i009:  Multi-system program queueing package, Part03/03
Newsgroups: comp.sources.unix
Approved: [e-mail]
X-Checksum-Snefru: 65243a6f 3368d383 5c915f62 48133fd4

Submitted-by: Scott Bradner <[e-mail]>
Posting-number: Volume 22, Issue 9
Archive-name: queuer/part03

#! /bin/sh
# This is a shell archive.  Remove anything before this line, then unpack
# it by saving it into a file and typing "sh file".  To overwrite existing
# files, type "sh file -c".  You can also feed this as standard input via
# unshar, or by typing "sh <file", e.g..  If this archive is complete, you
# will see the following message at the end:
#        "End of archive 3 (of 3)."
# Contents:  enqueue.c qmaster.c
# Wrapped by [e-mail] on Tue May  1 17:18:29 1990
PATH=/bin:/usr/bin:/usr/ucb ; export PATH
if test -f 'enqueue.c' -a "${1}" != "-c" ; then
  echo shar: Will not clobber existing file \"'enqueue.c'\"
else
echo shar: Extracting \"'enqueue.c'\" \(11727 characters\)
sed "s/^X//" >'enqueue.c' <<'END_OF_FILE'
X/* Copyright 1990  The President and Fellows of Harvard University
X
XPermission to use, copy, modify, and distribute this program for any
Xpurpose and without fee is hereby granted, provided that this
Xcopyright and permission notice appear on all copies and supporting
Xdocumentation, the name of Harvard University not be used in advertising
Xor publicity pertaining to distribution of the program, or to results
Xderived from its use, without specific prior written permission, and notice
Xbe given in supporting documentation that copying and distribution is by
Xpermission of Harvard University.  Harvard University makes no
Xrepresentations about the suitability of this software for any purpose.
XIt is provided "as is" without express or implied warranty.    */
X
X
X/* enqueue.c - Dan Lanciani '85 */
X
X#include <sys/param.h>
X#include <sys/stat.h>
X#include <sys/socket.h>
X#include <netinet/in.h>
X#include <sys/times.h>
X#include <signal.h>
X#include <netdb.h>
X#include <pwd.h>
X#include <grp.h>
X#include <errno.h>
X#include <stdio.h>
X
X#include "queue.h"
X
Xextern int errno;
Xextern char **environ;
Xint queuegroup, flat, pid, nofiles, nifiles, byebye();
Xtime_t starttime, time();
Xchar **ofiles, **ifiles, **lfiles;
Xlong unid;
X
Xenqueue(s, host, sin, local, recover)
Xchar *host;
Xstruct sockaddr_in *sin;
X{
X    int se = -1, sargc, i, uid, gid, ng, nenviron;
X    register long l;
X    gid_t groups[NGROUPS];
X    char **sargv, buf[BUFSIZ], *user, *group, *grl[XGROUPS];
X    char *cwd, *xcwd, **envp, name1[BUFSIZ], name2[BUFSIZ];
X    register struct passwd *pw;
X    register struct group *gr;
X    struct stat statb;
X    struct hostent *hp;
X    struct servent *sp;
X    struct sockaddr_in sin2;
X    register FILE *n, *m, *ck;
X    FILE *popen();
X
X    if(gr = getgrnam(QUEUEGROUP))
X        queuegroup = gr->gr_gid;
X    else
X        queuegroup = NOBODY;
X    getstr(s, buf);
X    if(recover) {
X        host = newstring(buf);
X        gethostname(buf, sizeof(buf));
X        local = !strcmp(host, buf);
X    }
X    else {
X        i = (u_short)atoi(buf);
X        if(i) {
X            if(i >= IPPORT_RESERVED)
X                exit(1);
X            ng = IPPORT_RESERVED - 1;
X            if((se = rresvport(&ng)) < 0)
X                exit(1);
X            sin->sin_port = htons((u_short)i);
X            if(connect(se, sin, sizeof(struct sockaddr_in)))
X                exit(1);
X        }
X    }
X    flat = local || infile(FLATFILE, host);
X    sprintf(buf, "%s/%d", SPOOLDIR, getpid());
X    if(!(ck = fopen(buf, "w")))
X        exit(1);
X    xfputs(host, ck);
X    getstr(s, buf);
X    unid = atol(buf);
X    sprintf(buf, "%s/%ld", SPOOLDIR, unid);
X    if(unid)
X        unlink(buf);
X    if(recover)
X        unid = 0;
X    fprintf(ck, "%ld", unid);
X    putc('\0', ck);
X    if(unid) {
X        if(!(n = fopen(buf, "w")))
X            exit(1);
X        fprintf(n, "%d", getpid());
X        putc('\0', n);
X        fclose(n);
X    }
X    getstr(s, buf);
X    xfputs(buf, ck);
X    sargc = atoi(buf);
X    sargv = (char **)malloc((sargc+1) * sizeof(char *));
X    for(i = 0; i < sargc; i++) {
X        sargv[i] = newstring(getstr(s, buf));
X        xfputs(buf, ck);
X    }
X    sargv[i] = NULL;
X    user = newstring(getstr(s, buf));
X    xfputs(buf, ck);
X    if(flat) {
X        if(!(pw = getpwnam(user)))
X            exit(1);
X        uid = pw->pw_uid;
X    }
X    else
X        uid = NOBODY;
X    group = newstring(getstr(s, buf));
X    xfputs(buf, ck);
X    if(flat) {
X        if(!(gr = getgrnam(group)))
X            exit(1);
X        gid = gr->gr_gid;
X    }
X    else
X        gid = NOBODY;
X    getstr(s, buf);
X    xfputs(buf, ck);
X    if((ng = atoi(buf)) > XGROUPS)
X        exit(1);
X    for(i = 0; i < ng; i++) {
X        grl[i] = newstring(getstr(s, buf));
X        xfputs(buf, ck);
X        if(flat) {
X            if(!(gr = getgrnam(grl[i])))
X                exit(1);
X            groups[i] = gr->gr_gid;
X        }
X        else
X            groups[i] = NOGROUP;
X    }
X    xcwd = cwd = newstring(getstr(s, buf));
X    xfputs(buf, ck);
X    getstr(s, buf);
X    nenviron = atoi(buf);
X    xfputs(buf, ck);
X    envp = (char **)malloc((nenviron+1) * sizeof(char *));
X    for(i = 0; i < nenviron; i++) {
X        envp[i] = newstring(getstr(s, buf));
X        xfputs(buf, ck);
X    }
X    envp[i] = NULL;
X    environ = envp;
X    if(readconf(sargv[0]))
X        exit(1);
X    if(recover) {
X        if(mode == QM_INTERACTIVE || !restart) {
X            fclose(ck);
X            sprintf(buf, "%s/%d", SPOOLDIR, getpid());
X            unlink(buf);
X            sprintf(buf, "%s/%d.dir", SPOOLDIR, getpid());
X            if(!access(buf, 0)) {
X                if(!vfork()) {
X                    execl("/bin/rm", "rm", "-rf", buf, 0);
X                    exit(1);
X                }
X                wait(0);
X            }
X            exit(0);
X        }
X        mode = QM_BATCH;
X    }
X
X    if(!local) {
X        if(!recover) {
X            sprintf(buf, "%s/%d.dir", SPOOLDIR, getpid());
X            if(!access(buf, 0)) {
X                if(!vfork()) {
X                    execl("/bin/rm", "rm", "-rf", buf, 0);
X                    exit(1);
X                }
X                wait(0);
X            }
X            mkdir(buf, 0700);
X        }
X        chown(buf, uid, gid);
X        xcwd = newstring(buf);
X    }
X    if(priv)
X        setregid(gid, queuegroup);
X    else
X        setgid(gid);
X    setgroups(ng, groups);
X    setreuid(uid, 0);
X    chdir(xcwd);
X    nofiles = nifiles = 0;
X    while(1) {
X        getstr(s, buf);
X        xfputs(buf, ck);
X        if(!strcmp(buf, "done"))
X            break;
X        if(!strcmp(buf, "copyout")) {
X            getstr(s, buf);
X            xfputs(buf, ck);
X            nofiles = atoi(buf);
X            ofiles = (char **)malloc(nofiles * sizeof(char *));
X            for(i = 0; i < nofiles; i++) {
X                getstr(s, buf);
X                xfputs(buf, ck);
X                ofiles[i] = newstring(buf);
X            }
X            continue;
X        }
X        if(!strcmp(buf, "efs")) {
X            getstr(s, buf);
X            xfputs(buf, ck);
X            nifiles = atoi(buf);
X            ifiles = (char **)malloc(nifiles * sizeof(char *));
X            lfiles = (char **)malloc(nifiles * sizeof(char *));
X            for(i = 0; i < nifiles; i++) {
X                getstr(s, buf);
X                xfputs(buf, ck);
X                ifiles[i] = newstring(buf);
X#ifdef    SANEEFS
X                if(ifiles[i][0] == '/') {
X                    sprintf(buf,"/r/%s%s",host,ifiles[i]);
X                    lfiles[i] = newstring(buf);
X                }
X                else
X                    lfiles[i] = ifiles[i];
X#else
X                getstr(s, buf);
X                xfputs(buf, ck);
X                lfiles[i] = newstring(buf);
X#endif
X            }
X            if(xcwd != cwd)
X                free(xcwd);
X            if(strncmp(cwd, "/r/", 3)) {
X                sprintf(buf, "/r/%s%s", host, cwd);
X                xcwd = newstring(buf);
X            }
X            else
X                xcwd = cwd;
X            chdir(xcwd);
X            continue;
X        }
X        if(!strcmp(buf, "copyin")) {
X            getstr(s, buf);
X            xfputs(buf, ck);
X            nifiles = atoi(buf);
X            ifiles = (char **)malloc(nifiles * sizeof(char *));
X            lfiles = (char **)malloc(nifiles * sizeof(char *));
X            for(i = 0; i < nifiles; i++) {
X                getstr(s, buf);
X                xfputs(buf, ck);
X                ifiles[i] = newstring(buf);
X                if(rindex(ifiles[i], '/'))
X                    lfiles[i] = rindex(ifiles[i], '/') + 1;
X                else
X                    lfiles[i] = ifiles[i];
X                if(!recover) {
X                    if(!(n = fopen(lfiles[i], "w")))
X                        exit(1);
X                    chown(lfiles[i], uid, gid);
X                    getstr(s, buf);
X                    l = atol(buf);
X                    if(l < 0) {
X                        unlink(lfiles[i]);
X                        l = 0;
X                    }
X                    while(l) {
X                        ng = sizeof(buf);
X                        if(ng > l)
X                            ng = l;
X                        if((ng = read(s, buf, ng)) <= 0)
X                            exit(1);
X                        l -= ng;
X                        fwrite(buf, ng, 1, n);
X                    }
X                    fclose(n);
X                }
X            }
X        continue;
X        }
X    }
X    fclose(ck);
X    for(i = 1; i < sargc; i++)
X        for(ng = 0; ng < nifiles; ng++)
X            if(!strcmp(sargv[i], ifiles[ng]))
X                sargv[i] = lfiles[ng];
X
X    if(mode == QM_BATCH) {
X        close(s);
X        if(se >= 0)
X            close(se);
X        sprintf(buf, "%s/%d.batch", SPOOLDIR, getpid());
X        s = creat(buf, 0600);
X        chown(buf, uid, gid);
X        sprintf(buf, "%s/%d.ebatch", SPOOLDIR, getpid());
X        se = creat(buf, 0600);
X        chown(buf, uid, gid);
X    }
X    for(i = 0; i < 3; i++)
X        dup2(s, i);
X    if(se >= 0)
X        dup2(se, 2);
X    close(s);
X    close(se);
X    if(mode != QM_INTERACTIVE) {
X        close(0);
X        open("/dev/null", 2);
X    }
X
X    waitrun(queue);
X    if(minload)
X        while(getload() > minload)
X            sleep(60);
X    sprintf(buf, "%s/%d", SPOOLDIR, getpid());
X    chmod(buf, 0755);
X    pid = 0;
X    starttime = time(0);
X    signal(SIGTERM, byebye);
X    pid = cspawn(prog, sargv);
X    pcontrol();
X    while((i = wait(0)) != pid)
X        if(i < 0 && errno != EINTR)
X            break;
X    signal(SIGALRM, SIG_IGN);
X    alarm(0);
X    signal(SIGTERM, SIG_IGN);
X    killpg(pid, SIGHUP);
X    killpg(pid, SIGCONT);
X    sleep(2);
X    killpg(pid, 9);
X    pid = getpid();
X    if(!access(QACCT, 0) && (n = fopen(QACCT, "a"))) {
X        struct tms tms;
X        times(&tms);
X        fprintf(n, "%ld\t%ld\t%s\t%s\t%s\t%s\n",
X            tms.tms_utime, tms.tms_stime,
X            sargv[0], user, group, host);
X        fclose(n);
X    }
X    if(!local) {
X        sprintf(buf, "%s/%d.dir", SPOOLDIR, pid);
X        if(!access(buf, 0)) {
X            if(!vfork()) {
X                execl("/bin/rm", "rm", "-rf", buf, 0);
X                exit(1);
X            }
X            wait(0);
X        }
X    }
X    if(unid) {
X        sprintf(buf, "%s/%ld", SPOOLDIR, unid);
X        unlink(buf);
X    }
X    sprintf(name1, "%s/%s", SPOOLDIR, queue);
X    lock(name1);
X    n = fopen(name1, "r");
X    strcpy(name2, name1);
X    strcat(name2, ".tmp");
X    m = fopen(name2, "w");
X    i = 0;
X    while(fgets(buf, sizeof(buf), n))
X        if(atoi(buf) != pid) {
X            if(i++ < maxrun)
X                kill(atoi(buf), SIGALRM);
X            fputs(buf, m);
X        }
X    fclose(m);
X    fclose(n);
X    unlink(name1);
X    link(name2, name1);
X    unlink(name2);
X    unlock(name1);
X    for(i = 0; i < 3; i++)
X        close(i);
X    open("/dev/null", 2);
X    dup(0);
X    dup(0);
X    if(*qm && (sp = getservbyname("qmaster", "udp")) &&
X        (hp = gethostbyname(qm)) &&
X        (i = socket(AF_INET, SOCK_DGRAM, 0)) >= 0) {
X        *buf = 0;
X        if(unid)
X            sprintf(buf + 1, "%ld", unid);
X        else
X            sprintf(buf + 1, "%d", pid);
X        sin2.sin_family = hp->h_addrtype;
X        sin2.sin_port = sp->s_port;
X        bcopy(hp->h_addr, &sin2.sin_addr, hp->h_length);
X        sendto(i, buf, 2 + strlen(buf + 1), 0, &sin2, sizeof(sin2));
X        close(i);
X    }
X    if(mode == QM_BATCH) {
X        int sentsome = 0;
X        char *p;
X
X        sprintf(name1, "%s@%s", user, host);
X        name2[0] = '\0';
X        for(i = 0; i < sargc - 1; i++) {
X            strcat(name2, sargv[i]);
X            strcat(name2, " ");
X        }
X        strcat(name2, sargv[sargc - 1]);
X        sprintf(buf, "%s/%d.batch", SPOOLDIR, pid);
X        close(0);
X        open(buf, 0);
X        unlink(buf);
X        if(!fstat(0, &statb) && statb.st_size) {
X            sprintf(buf, "batch job %d output (%s)", pid, name2);
X            if(!vfork()) {
X                setuid(uid);
X                execl("/usr/ucb/Mail", "Mail", "-s",
X                    buf, name1, 0);
X                exit(1);
X            }
X            wait(0);
X            sentsome++;
X        }
X        sprintf(buf, "%s/%d.ebatch", SPOOLDIR, pid);
X        close(0);
X        open(buf, 0);
X        unlink(buf);
X        if(!fstat(0, &statb) && statb.st_size) {
X            sprintf(buf, "batch job %d errors (%s)", pid, name2);
X            if(!vfork()) {
X                setuid(uid);
X                execl("/usr/ucb/Mail", "Mail", "-s",
X                    buf, name1, 0);
X                exit(1);
X            }
X            wait(0);
X            sentsome++;
X        }
X        close(0);
X        dup(1);
X        sprintf(buf, "%s/%d", SPOOLDIR, pid);
X        unlink(buf);
X        setuid(uid);
X        if(!(p = getenv("QNOTIFY")))
X            p = "mail";
X        if(!strcmp(p, "send") || !strcmp(p,"saml")||!strcmp(p,"soml")) {
X            sprintf(buf, "exec /usr/lib/sendmail -S -eq %s", name1);
X            i = 1;
X            if(n = popen(buf, "w")) {
X            fprintf(n, "Your batch job %d is finished.\n", pid);
X            fprintf(n, "\"%s\"\n", name2);
X            i = pclose(n);
X            }
X        }
X        if(!sentsome &&
X(!strcmp(p, "mail") || !strcmp(p, "saml") || (!strcmp(p, "soml") && i))) {
X            sprintf(buf, "exec /usr/ucb/Mail -s 'batch job' %s",
X                name1);
X            if(n = popen(buf, "w")) {
X            fprintf(n, "Your batch job %d is finished.\n", pid);
X            fprintf(n, "\"%s\"\n", name2);
X            pclose(n);
X            }
X        }
X    }
X    else {
X        sprintf(buf, "%s/%d", SPOOLDIR, pid);
X        unlink(buf);
X    }
X}
X
Xbyebye()
X{
X    if(pid) {
X        killpg(pid, SIGHUP);
X        killpg(pid, SIGCONT);
X        kill(pid, SIGCONT);
X        killpg(pid, 9);
X        kill(pid, 9);
X    }
X    signal(SIGTERM, byebye);
X}
X
Xcatch()
X{
X    signal(SIGALRM, catch);
X}
X
Xint running = 1;
X
Xpcontrol()
X{
X    int load;
X
X    if(maxtime && time(0) - starttime > maxtime)
X        byebye();
X    if(minload != maxload) {
X        load = getload();
X        if(running) {
X            if(load >= maxload) {
X                running = 0;
X                killpg(pid, SIGSTOP);
X            }
X        }
X        else {
X            if(load <= minload) {
X                running = 1;
X                killpg(pid, SIGCONT);
X            }
X        }
X    }
X    signal(SIGALRM, pcontrol);
X    alarm(60);
X}
X
Xwaitrun(q)
Xchar *q;
X{
X    int i, pid;
X    char buf[BUFSIZ], p[BUFSIZ];
X    FILE *n;
X
X    signal(SIGALRM, catch);
X    sprintf(p, "%s/%s", SPOOLDIR, q);
X    pid = getpid();
X    lock(p);
X    if(!(n = fopen(p, "a"))) {
X        unlock(p);
X        exit(1);
X    }
X    fprintf(n, "%d\n", pid);
X    fclose(n);
X    unlock(p);
X    while(1) {
X        lock(p);
X        if(!(n = fopen(p, "r"))) {
X            unlock(p);
X            exit(1);
X        }
X        for(i = 0; i < maxrun; i++) {
X            fgets(buf, sizeof(buf), n);
X            buf[strlen(buf)-1] = '\0';
X            if(atoi(buf) == pid) {
X                fclose(n);
X                unlock(p);
X                return;
X            }
X        }
X        fclose(n);
X        unlock(p);
X        alarm(300);
X        pause();
X        alarm(0);
X    }
X}
END_OF_FILE
if test 11727 -ne `wc -c <'enqueue.c'`; then
    echo shar: \"'enqueue.c'\" unpacked with wrong size!
fi
# end of 'enqueue.c'
fi
if test -f 'qmaster.c' -a "${1}" != "-c" ; then
  echo shar: Will not clobber existing file \"'qmaster.c'\"
else
echo shar: Extracting \"'qmaster.c'\" \(19051 characters\)
sed "s/^X//" >'qmaster.c' <<'END_OF_FILE'
X/* Copyright 1990  The President and Fellows of Harvard University
X
XPermission to use, copy, modify, and distribute this program for any
Xpurpose and without fee is hereby granted, provided that this
Xcopyright and permission notice appear on all copies and supporting
Xdocumentation, the name of Harvard University not be used in advertising
Xor publicity pertaining to distribution of the program, or to results
Xderived from its use, without specific prior written permission, and notice
Xbe given in supporting documentation that copying and distribution is by
Xpermission of Harvard University.  Harvard University makes no
Xrepresentations about the suitability of this software for any purpose.
XIt is provided "as is" without express or implied warranty.    */
X
X
X/* qmaster.c - Dan Lanciani '89 */
X
X/*
Xrefresh hosts less
Xpick up strays
X*/
X
X#include <stdio.h>
X#include <sys/types.h>
X#include <sys/socket.h>
X#include <sys/wait.h>
X#include <sys/time.h>
X#include <netinet/in.h>
X#include <netdb.h>
X#include <errno.h>
X#include <signal.h>
X#include <sgtty.h>
X#include <setjmp.h>
X
X#include "queue.h"
X
X#define GARBAGE (1*60)
X#define HOLDDOWN (2*60)
X#define HOSTPOLLTIME (5*60)
X#define DEADMAN (1*60)
X
Xstruct hostinfo {
X    struct hostinfo *hi_next;
X    char *hi_name;
X    struct sockaddr_in hi_addr;
X    int hi_equiv;
X    int hi_load;
X    time_t hi_timestamp;
X    int hi_dead;
X    int hi_queued;
X} *hostinfo;
X
Xstruct userinfo {
X    struct userinfo *ui_next;
X    long ui_unid;
X    struct sockaddr_in ui_addr;
X    char *ui_name;
X    struct proginfo *ui_prog;
X    time_t ui_timestamp;
X    int ui_mark;
X};
X
Xstruct queueinfo {
X    struct queueinfo *qi_next;
X    char *qi_name;
X    int qi_maxrun;
X    int qi_minload;
X    int qi_maxperu;
X    int qi_hcnt;
X    struct hostinfo *qi_hosts[MAXHCNT];
X    struct userinfo *qi_heads[MAXHCNT];
X    struct userinfo *qi_head;
X} *queueinfo;
X
Xstruct proginfo {
X    struct proginfo *pi_next;
X    char *pi_name;
X    struct queueinfo *pi_queue;
X} *proginfo;
X
Xstruct sockaddr_in sin = { AF_INET };
Xtime_t time();
Xextern int errno;
Xint reapchild();
Xlong atol();
X
Xlong
Xnewunid()
X{
X    static int fd = -1;
X    long unid;
X    char buf[100];
X
X    if(fd < 0) {
X        if((fd = open(UNID, 2)) < 0) {
X            if((fd = creat(UNID, 0644)) < 0) {
X                perror("creat");
X                exit(1);
X            }
X            sprintf(buf, "%ld", FIRSTUNID);
X            write(fd, buf, strlen(buf) + 1);
X            close(fd);
X            if((fd = open(UNID, 2)) < 0) {
X                perror("open");
X                exit(1);
X            }
X        }
X    }
X    lseek(fd, 0L, 0);
X    read(fd, buf, sizeof(buf));
X    unid = atol(buf) + 1;
X    if(!isunid(unid))
X        unid = FIRSTUNID;
X    sprintf(buf, "%ld", unid);
X    lseek(fd, 0L, 0);
X    write(fd, buf, strlen(buf) + 1);
X    return(unid);
X}
X
X#define refresh(hi) \
X    if(time((time_t)0) - (hi)->hi_timestamp > HOSTPOLLTIME) { \
X        (hi)->hi_load = getrload((hi)->hi_name); \
X        (hi)->hi_timestamp = time((time_t)0); \
X    } \
X    else
X
Xstruct hostinfo *
Xhibyname(name)
Xregister char *name;
X{
X    register struct hostinfo *hi;
X    register struct hostent *hp;
X
X    for(hi = hostinfo; hi; hi = hi->hi_next)
X        if(!strcmp(hi->hi_name, name)) {
X            refresh(hi);
X            return(hi);
X        }
X    if(!(hp = gethostbyname(name)))
X        return(0);
X    hi = (struct hostinfo *)malloc(sizeof(struct hostinfo));
X    hi->hi_name = newstring(name);
X    if(infile(EQUIVFILE, hi->hi_name))
X        hi->hi_equiv = 1;
X    else
X        hi->hi_equiv = 0;
X    hi->hi_addr.sin_family = hp->h_addrtype;
X    bcopy(hp->h_addr, &hi->hi_addr.sin_addr, hp->h_length);
X    hi->hi_load = getrload(name);
X    hi->hi_timestamp = time((time_t)0);
X    hi->hi_dead = 0;
X    hi->hi_queued = 0;
X    hi->hi_next = hostinfo;
X    hostinfo = hi;
X    return(hi);
X}
X
Xstruct hostinfo *
Xhibyaddr(addr)
Xregister struct sockaddr_in *addr;
X{
X    register struct hostinfo *hi;
X    register struct hostent *hp;
X
X    for(hi = hostinfo; hi; hi = hi->hi_next)
X        if(addr->sin_addr.s_addr == hi->hi_addr.sin_addr.s_addr) {
X            refresh(hi);
X            return(hi);
X        }
X    if(!(hp=gethostbyaddr(&addr->sin_addr, sizeof(struct in_addr),AF_INET)))
X        return(0);
X    hi = (struct hostinfo *)malloc(sizeof(struct hostinfo));
X    hi->hi_name = newstring(hp->h_name);
X    if(index(hi->hi_name, '.'))
X        *index(hi->hi_name, '.') = '\0';
X    if(infile(EQUIVFILE, hi->hi_name))
X        hi->hi_equiv = 1;
X    else
X        hi->hi_equiv = 0;
X    hi->hi_addr.sin_family = hp->h_addrtype;
X    bcopy(hp->h_addr, &hi->hi_addr.sin_addr, hp->h_length);
X    hi->hi_load = getrload(hi->hi_name);
X    hi->hi_timestamp = time((time_t)0);
X    hi->hi_dead = 0;
X    hi->hi_queued = 0;
X    hi->hi_next = hostinfo;
X    hostinfo = hi;
X    return(hi);
X}
X
Xstruct userinfo *
Xuibyunid(unid, unlnk)
Xregister long unid;
X{
X    register struct queueinfo *qi;
X    register struct userinfo *ui, *pu;
X    register int i;
X
X    for(qi = queueinfo; qi; qi = qi->qi_next) {
X        for(i = 0; i < qi->qi_hcnt; i++)
X            for(pu = 0, ui =qi->qi_heads[i];ui;pu=ui,ui=ui->ui_next)
X                if(unid == ui->ui_unid) {
X                    if(unlnk)
X                    if(pu)
X                        pu->ui_next = ui->ui_next;
X                    else
X                        qi->qi_heads[i] = ui->ui_next;
X                    return(ui);
X                }
X        for(pu = 0, ui = qi->qi_head; ui; pu = ui, ui = ui->ui_next)
X            if(unid == ui->ui_unid) {
X                if(unlnk)
X                    if(pu)
X                        pu->ui_next = ui->ui_next;
X                    else
X                        qi->qi_head = ui->ui_next;
X                return(ui);
X            }
X    }
X    return(0);
X}
X
Xstruct queueinfo *
Xqibyname(name)
Xregister char *name;
X{
X    register struct queueinfo *qi;
X
X    for(qi = queueinfo; qi; qi = qi->qi_next)
X        if(!strcmp(qi->qi_name, name))
X            return(qi);
X    return(0);
X}
X
Xstruct proginfo *
Xpibyname(name)
Xregister char *name;
X{
X    register struct proginfo *pi;
X    register struct queueinfo *qi;
X    register int i;
X
X    for(pi = proginfo; pi; pi = pi->pi_next)
X        if(!strcmp(pi->pi_name, name))
X            return(pi);
X    if(readconf(name))
X        return(0);
X    if(!(qi = qibyname(queue))) {
X        qi = (struct queueinfo *)malloc(sizeof(struct queueinfo));
X        qi->qi_name = newstring(queue);
X        qi->qi_maxrun = maxrun;
X        qi->qi_minload = minload;
X        qi->qi_maxperu = maxperu;
X        for(i = 0; i < hcnt; i++) {
X            if(!(qi->qi_hosts[i] = hibyname(hosts[i]))) {
X                i--;
X                hcnt--;
X                continue;
X            }
X            qi->qi_hosts[i]->hi_queued = 1;
X            qi->qi_heads[i] = 0;
X        }
X        if(hcnt <= 0) {
X            free(qi);
X            return(0);
X        }
X        qi->qi_hcnt = hcnt;
X        qi->qi_head = 0;
X        qi->qi_next = queueinfo;
X        queueinfo = qi;
X    }
X    pi = (struct proginfo *)malloc(sizeof(struct proginfo));
X    pi->pi_name = newstring(name);
X    pi->pi_queue = qi;
X    pi->pi_next = proginfo;
X    proginfo = pi;
X    return(pi);
X}
X
Xchar *
Xgetln(n, buf)
Xchar *buf;
X{
X    register char *p = buf;
X
X    do
X        if(read(n, p, 1) != 1)
X            return(0);
X    while(*p++ != '\n');
X    p[-1] = 0;
X    return(buf);
X}
X
Xchar *
Xgetstr(n, buf)
Xchar *buf;
X{
X    register char *p = buf;
X
X    do
X        if(read(n, p, 1) != 1)
X            return(0);
X    while(*p++);
X    return(buf);
X}
X
Xstruct timeval timeout = { 60, 0 };
Xchar myname[100];
Xint s, u, on = 1;
Xint strays, dropouts, upstarts;
Xint deadmans, enqueues, dequeues, showqueues;
Xjmp_buf jb;
X
Xcatch()
X{
X    deadmans++;
X    longjmp(jb, 1);
X}
X
Xmain(argc, argv, envp)
Xchar **argv, **envp;
X{
X    register struct servent *sp;
X    register struct queueinfo *qi;
X    register struct userinfo *ui, *pu;
X    register struct hostinfo *hi, *garbagehi = 0;
X    register int i, j, qport;
X    long probe, mask, unid;
X    char buf[BUFSIZ];
X    time_t garbage = 0;
X
X    gethostname(myname, sizeof(myname));
X    if(argc > 1) {
X        for(i = 1; i < argc; i++)
X            if(!strcmp(argv[i], myname))
X                goto iammaster;
Xfprintf(stderr, "I am not master %s\n", myname);
X        exit(0);
X    }
Xiammaster:
X#ifndef DEBUG
X    for(s = 0; s < 30; s++)
X        close(s);
X    if((s = open("/dev/tty", 2)) >= 0) {
X        ioctl(s, TIOCNOTTY, 0);
X        close(s);
X    }
X    s = open("/", 0);
X    dup(s);
X    dup(s);
X    signal(SIGALRM, SIG_IGN);
X    signal(SIGPIPE, SIG_IGN);
X    if(fork())
X        exit(0);
X#endif
X
X    while(fork()) {
X        wait((int *)0);
X        sleep(15);
X    }
X    if(!(sp = getservbyname("queue", "tcp"))) {
X        fprintf(stderr, "queue/tcp: Bad service?!?\n");
X        exit(1);
X    }
X    qport = sp->s_port;
X    if(!(sp = getservbyname("qmaster", "tcp"))) {
X        fprintf(stderr, "qmaster/tcp: Bad service?!?\n");
X        exit(1);
X    }
X    if((s = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
X        perror("socket");
X        exit(1);
X    }
X    setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
X    setsockopt(s, SOL_SOCKET, SO_KEEPALIVE, &on, sizeof(on));
X    sin.sin_port = sp->s_port;
X    if(bind(s, &sin, sizeof(sin))) {
X        perror("bind");
X        exit(1);
X    }
X    listen(s, 10);
X    mask = (1L << s);
X    if(!(sp = getservbyname("qmaster", "udp"))) {
X        fprintf(stderr, "qmaster/udp: Bad service?!?\n");
X        exit(1);
X    }
X    if((u = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
X        perror("socket");
X        exit(1);
X    }
X    setsockopt(u, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
X    sin.sin_port = sp->s_port;
X    if(bind(u, &sin, sizeof(sin))) {
X        perror("bind");
X        exit(1);
X    }
X    mask |= (1L << u);
X    while(1) {
X        struct sockaddr_in from;
X        int s0, fromlen;
X        signal(SIGCHLD, reapchild);
X        probe = mask;
X        if(select(32, &probe, 0, 0, &timeout) < 0) {
X            sleep(1);
X            continue;
X        }
X        if(probe & (1L << s)) {
X            fromlen = sizeof(from);
X            if((s0 = accept(s, &from, &fromlen)) < 0) {
X                if(errno = EINTR)
X                    continue;
X                perror("accept");
X                sleep(1);
X                continue;
X            }
X            if(from.sin_family != AF_INET ||
X                htons((u_short)from.sin_port) >= IPPORT_RESERVED
X                || htons((u_short)from.sin_port)
X                < IPPORT_RESERVED / 2) {
X                close(s0);
X                continue;
X            }
X            dostream(s0, &from);
X        }
X        if(probe & (1L << u))
X            dodgram(u);
X        for(hi = hostinfo; hi; hi = hi->hi_next)
X            if(hi->hi_queued)
X                refresh(hi);
X        for(qi = queueinfo; qi; qi = qi->qi_next) {
X            if(!qi->qi_head)
X                continue;
X            for(i = 0; i < qi->qi_hcnt; i++)
X                for(j = i + 1; j < qi->qi_hcnt; j++)
X                    if(qi->qi_hosts[j]->hi_load <
X                        qi->qi_hosts[i]->hi_load) {
X                        ui = qi->qi_heads[i];
X                        hi = qi->qi_hosts[i];
X                        qi->qi_heads[i]=qi->qi_heads[j];
X                        qi->qi_hosts[i]=qi->qi_hosts[j];
X                        qi->qi_heads[j] = ui;
X                        qi->qi_hosts[j] = hi;
X                    }
X            for(i = 0; i < qi->qi_hcnt && qi->qi_head; i++) {
X                if(qi->qi_hosts[i]->hi_dead)
X                    continue;
X                if(qi->qi_minload &&
X                    qi->qi_hosts[i]->hi_load>qi->qi_minload)
X                    continue;
X                j = 0;
X                for(ui = qi->qi_heads[i]; ui; ui = ui->ui_next)
X                    j++;
X                while(j < qi->qi_maxrun && (ui = qi->qi_head)) {
X                    qi->qi_head = ui->ui_next;
X                    ui->ui_next = qi->qi_heads[i];
X                    qi->qi_heads[i] = ui;
X                    sprintf(buf, "\1%s",
X                        qi->qi_hosts[i]->hi_name);
X                    sendto(u, buf, strlen(buf) + 1, 0,
X                        &ui->ui_addr,
X                        sizeof(ui->ui_addr));
X                    ui->ui_timestamp = time((time_t)0);
X                    j++;
X                }
X            }
X            *buf = 0;
X            for(ui = qi->qi_head; ui; ui = ui->ui_next)
X                sendto(u, buf, 1, 0, &ui->ui_addr,
X                    sizeof(ui->ui_addr));
X        }
X        if(time((time_t)0) - garbage > GARBAGE) {
X            int rp = IPPORT_RESERVED - 1;
X            register char *p;
X            if(!garbagehi)
X                garbagehi = hostinfo;
X            if(!(hi = garbagehi) || !hi->hi_queued)
X                goto out;
X            hi->hi_addr.sin_port = qport;
X            if((i = rresvport(&rp)) < 0)
X                goto out;
X            if(setjmp(jb))
X                goto out2;
X#ifdef DEBUG
X            fprintf(stderr, "gc host %s\n", hi->hi_name);
X#endif
X            signal(SIGALRM, catch);
X            alarm(DEADMAN);
X            if(connect(i, &hi->hi_addr, sizeof(hi->hi_addr))) {
X                hi->hi_dead = 1;
X                goto out2;
X            }
X            hi->hi_dead = 0;
X            alarm(DEADMAN);
X            write(i, "\1", 2);
X            alarm(DEADMAN);
X            while(getln(i, buf)) {
X                alarm(0);
X                if(strncmp(buf, "Queue: ", 7))
X                    goto out2;
X                if(!(p = index(buf + 7, ',')))
X                    goto out2;
X                *p = 0;
X#ifdef DEBUG
X                fprintf(stderr, "queue = %s\n", buf);
X#endif
X                if(!(qi = qibyname(buf + 7))) {
X                skip:
X                    alarm(DEADMAN);
X                    while(getln(i, buf) && *buf)
X                        alarm(DEADMAN);
X                    continue;
X                }
X                for(j = 0; j < qi->qi_hcnt; j++)
X                    if(hi == qi->qi_hosts[j])
X                        break;
X                if(j >= qi->qi_hcnt)
X                    goto skip;
X                for(ui = qi->qi_heads[j]; ui; ui = ui->ui_next)
X                    ui->ui_mark = 0;
X                alarm(DEADMAN);
X                if(!getln(i, buf) || strncmp(buf, "Pid", 3))
X                    goto out2;
X#ifdef DEBUG
X                fprintf(stderr, "header = %s\n", buf);
X#endif
X                while(getln(i, buf) && *buf) {
X                    alarm(0);
X#ifdef DEBUG
X                fprintf(stderr, "entry = %s\n", buf);
X#endif
X                    unid = atol(buf);
X                    for(ui = qi->qi_heads[j]; ui;
X                        ui = ui->ui_next)
X                        if(ui->ui_unid == unid) {
X                            /* XXX check mark */
X                            ui->ui_mark = 1;
X                            break;
X                        }
X                    if(ui) {
X                        alarm(DEADMAN);
X                        continue;
X                    }
X                    if(ui = uibyunid(unid, 1)) {
X                        ui->ui_mark = 1;
X                        ui->ui_next = qi->qi_heads[j];
X                        qi->qi_heads[j] = ui;
X                        upstarts++;
X                        alarm(DEADMAN);
X                        continue;
X                    }
X                    /* XXX add it */
X                    strays++;
X                    alarm(DEADMAN);
X                }
X                alarm(0);
X            again:
X                for(pu = 0, ui = qi->qi_heads[j]; ui;
X                pu = ui, ui = ui->ui_next)
X                if(!ui->ui_mark && time((time_t)0) -
X                    ui->ui_timestamp > HOLDDOWN) {
X                    if(pu)
X                        pu->ui_next = ui->ui_next;
X                    else
X                        qi->qi_heads[j] = ui->ui_next;
X                    free(ui->ui_name);
X                    free(ui);
X                    dropouts++;
X                    goto again;
X                }
X                alarm(DEADMAN);
X            }
X        out2:
X            close(i);
X        out:
X            alarm(0);
X            signal(SIGALRM, SIG_DFL);
X            garbagehi = hi->hi_next;
X            garbage = time((time_t)0);
X        }
X    }
X}
X
Xreapchild()
X{
X    union wait status;
X
X    while(wait3(&status, WNOHANG, 0) > 0);
X}
X
Xdostream(s, sin)
Xregister int s;
Xstruct sockaddr_in *sin;
X{
X    register struct hostinfo *hi;
X
X    if(!(hi = hibyaddr(sin)))
X        goto bad;
X    if(setjmp(jb))
X        goto bad;
X    signal(SIGALRM, catch);
X    alarm(DEADMAN);
X    if(!strcmp(hi->hi_name, "localhost") || !strcmp(hi->hi_name, myname))
X        qservice(s, myname, sin, 1);
X    else if(hi->hi_equiv)
X        qservice(s, hi->hi_name, sin, 0);
Xbad:
X    alarm(0);
X    signal(SIGALRM, SIG_IGN);
X    close(s);
X}
X
Xdodgram(s)
Xregister int s;
X{
X    char buf[BUFSIZ];
X    struct sockaddr_in sin;
X    int len, fromlen = sizeof(sin);
X    register struct userinfo *ui;
X
X    if(setjmp(jb))
X        goto bad;
X    signal(SIGALRM, catch);
X    alarm(10);
X    if((len = recvfrom(s, buf, sizeof(buf), 0, &sin, sizeof(sin))) <= 0)
X        goto bad;
X    alarm(0);
X    switch(*buf&0377) {
X
X        case 0:
X            if(ui = uibyunid(atol(buf + 1), 1)) {
X                free(ui->ui_name);
X                free(ui);
X            }
X            break;
X    }
Xbad:
X    alarm(0);
X    signal(SIGALRM, SIG_IGN);
X}
X
Xqservice(s, host, sin, local)
Xregister int s;
Xregister char *host;
Xregister struct sockaddr_in *sin;
Xregister int local;
X{
X    char request;
X
X    if(read(s, &request, 1) != 1)
X        return;
X    switch(request&0377) {
X
X        case 0:
X            menqueue(s, host, sin, local);
X            enqueues++;
X            break;
X
X        case 1:
X            mshowqueue(s, host, sin, local);
X            showqueues++;
X            break;
X
X        case 2:
X            mdequeue(s, host, sin, local);
X            dequeues++;
X            break;
X    }
X}
X
Xmenqueue(s, host, sin, local)
Xregister int s;
Xregister char *host;
Xregister struct sockaddr_in *sin;
Xregister int local;
X{
X    char what[BUFSIZ], prog[BUFSIZ], user[BUFSIZ], buf[BUFSIZ];
X    register struct proginfo *pi;
X    register struct queueinfo *qi;
X    register struct userinfo *ui, *u2;
X    register int i, j;
X
X    if(!getstr(s, what) || !getstr(s, prog) || !getstr(s, user))
X        return;
X    alarm(0);
X    if(!(pi = pibyname(prog)))
X        return;
X    qi = pi->pi_queue;
X    if(qi->qi_maxperu) {
X        for(i = j = 0; i < qi->qi_hcnt; i++)
X            for(ui = qi->qi_heads[i]; ui; ui = ui->ui_next)
X                if(!strcmp(user, ui->ui_name))
X                    j++;
X        for(ui = qi->qi_head; ui; ui = ui->ui_next)
X            if(!strcmp(user, ui->ui_name))
X                j++;
X        if(j >= qi->qi_maxperu) {
X            *buf = 1;
X            sprintf(buf + 1, "You already have %d jobs queued", j);
X            alarm(DEADMAN);
X            write(s, buf, strlen(buf + 1) + 2);
X            return;
X        }
X    }
X    ui = (struct userinfo *)malloc(sizeof(struct userinfo));
X    *buf = 0;
X    sprintf(buf + 1, "%ld", ui->ui_unid = newunid());
X    alarm(DEADMAN);
X    write(s, buf, strlen(buf + 1) + 2);
X    alarm(0);
X    ui->ui_addr = *sin;
X    ui->ui_addr.sin_port = atoi(what);
X    ui->ui_name = newstring(user);
X    ui->ui_prog = pi;
X    ui->ui_next = 0;
X    if(u2 = qi->qi_head) {
X        while(u2->ui_next)
X            u2 = u2->ui_next;
X        u2->ui_next = ui;
X    }
X    else
X        qi->qi_head = ui;
X}
X
Xmshowqueue(s, host, sin, local)
Xregister int s;
Xregister char *host;
Xregister struct sockaddr_in *sin;
Xregister int local;
X{
X    register int i;
X    register struct queueinfo *qi;
X    register struct userinfo *ui;
X    register struct hostinfo *hi;
X    char buf[BUFSIZ];
X
X    alarm(0);
X    if(fork())
X        return;
X    for(i = 0; i < 3; i++)
X        dup2(s, i);
X    if(!getstr(s, buf))
X        exit(1);
X    if(s > 2)
X        close(s);
X    if(!strcmp(buf, "test")) {
X        printf("Debugging information:\n");
X        printf("Strays: %d, Dropouts: %d, Upstarts: %d\n",
X            strays, dropouts, upstarts);
X        printf("Deadmans: %d\n", deadmans);
X        printf("Enqueues: %d, Dequeues: %d, Showqueues: %d\n",
X                enqueues, dequeues, showqueues);
X        printf("Host\t\tAddr\t\tEquiv\tLoad\tDead\tQueued\n");
X        for(hi = hostinfo; hi; hi = hi->hi_next)
X            printf("%-16s%s\t%d\t%d\t%d\t%d\n",
X                hi->hi_name, inet_ntoa(hi->hi_addr.sin_addr),
X                hi->hi_equiv, hi->hi_load,
X                hi->hi_dead, hi->hi_queued);
X        printf("\n");
X    }
X    printf("Queue: %s\n", buf);
X    printf("Pid\t\tState\tUser\tHost\tCommand\n");
X    fflush(stdout);
X    if(!(qi = qibyname(buf)))
X        exit(1);
X    for(i = 0; i < qi->qi_hcnt; i++)
X        for(ui = qi->qi_heads[i]; ui; ui = ui->ui_next) {
X            printf("%-16ld", ui->ui_unid);
X            printf("RUN\t");
X            printf("%s\t", ui->ui_name);
X            if(hi = hibyaddr(&ui->ui_addr))
X                printf("%s\t", hi->hi_name);
X            else
X            printf("%s\t", inet_ntoa(ui->ui_addr.sin_addr.s_addr));
X            printf("%s\n", ui->ui_prog->pi_name);
X        }
X    for(ui = qi->qi_head; ui; ui = ui->ui_next) {
X        printf("%-16ld", ui->ui_unid);
X        printf("WAIT\t");
X        printf("%s\t", ui->ui_name);
X        if(hi = hibyaddr(&ui->ui_addr))
X            printf("%s\t", hi->hi_name);
X        else
X        printf("%s\t", inet_ntoa(ui->ui_addr.sin_addr.s_addr));
X        printf("%s\n", ui->ui_prog->pi_name);
X    }
X    fflush(stdout);
X    exit(0);
X}
X
Xmdequeue(s, host, sin, local)
Xregister int s;
Xregister char *host;
Xregister struct sockaddr_in *sin;
Xregister int local;
X{
X    char what[BUFSIZ], prog[BUFSIZ], user[BUFSIZ], buf[BUFSIZ];
X    long unid;
X    register struct proginfo *pi;
X    register struct queueinfo *qi;
X    register struct userinfo *ui, *pu;
X    register int i;
X
X    if(!getstr(s, user) || !getstr(s, prog) || !getstr(s, what))
X        return;
X    if(!(pi = pibyname(prog)))
X        return;
X    qi = pi->pi_queue;
X    if(strcmp(what, "all"))
X        unid = atol(what);
X    else
X        unid = -1;
X    alarm(DEADMAN);
X    for(i = 0; i < qi->qi_hcnt; i++)
X    top:    for(pu = 0, ui = qi->qi_heads[i]; ui; pu = ui, ui = ui->ui_next)
X            if(ui->ui_unid == unid ||
X                (!unid && !strcmp(ui->ui_name, what)) ||
X                unid == -1) {
X                if(strcmp(ui->ui_name, user) &&
X                    strcmp(user, "root")) {
X                    /*sprintf(buf, "Not owner (%s, %d)\n",
X                        ui->ui_name, ui->ui_unid);
X                    write(s, buf, strlen(buf));*/
X                    continue;
X                }
X                alarm(0);
X                if(pu)
X                    pu->ui_next = ui->ui_next;
X                else
X                    qi->qi_heads[i] = ui->ui_next;
X                free(ui->ui_name);
X                free(ui);
X                alarm(DEADMAN);
X                /*sprintf(buf, "Killed %d of %s for %s\n",
X                    ui->ui_unid, ui->ui_name, user);
X                write(s, buf, strlen(buf));*/
X                goto top;
X            }
X    alarm(DEADMAN);
Xtop2:    for(pu = 0, ui = qi->qi_head; ui; pu = ui, ui = ui->ui_next)
X        if(ui->ui_unid == unid ||
X            (!unid && !strcmp(ui->ui_name, what)) || unid == -1) {
X            if(strcmp(ui->ui_name, user) &&
X                strcmp(user, "root")) {
X                sprintf(buf, "Not owner (%s, %d)\n",
X                    ui->ui_name, ui->ui_unid);
X                write(s, buf, strlen(buf));
X                continue;
X            }
X            alarm(0);
X            if(pu)
X                pu->ui_next = ui->ui_next;
X            else
X                qi->qi_head = ui->ui_next;
X            *buf = 2;
X            sendto(u, buf, 1, 0, &ui->ui_addr, sizeof(ui->ui_addr));
X            free(ui->ui_name);
X            free(ui);
X            alarm(DEADMAN);
X            sprintf(buf, "Killed %d of %s for %s\n",
X                ui->ui_unid, ui->ui_name, user);
X            write(s, buf, strlen(buf));
X            goto top2;
X        }
X}
END_OF_FILE
if test 19051 -ne `wc -c <'qmaster.c'`; then
    echo shar: \"'qmaster.c'\" unpacked with wrong size!
fi
# end of 'qmaster.c'
fi
echo shar: End of archive 3 \(of 3\).
cp /dev/null ark3isdone
MISSING=""
for I in 1 2 3 ; do
    if test ! -f ark${I}isdone ; then
    MISSING="${MISSING} ${I}"
    fi
done
if test "${MISSING}" = "" ; then
    echo You have unpacked all 3 archives.
    rm -f ark[1-9]isdone
else
    echo You still need to unpack the following archives:
    echo "        " ${MISSING}
fi
##  End of shell archive.
exit 0
exit 0 # Just in case...